#include "Threadpool.h"
#include <assert.h>

thread_pool::thread_pool(int init_size)
	:m_init_threads_size(init_size),
	m_mutex(),
	m_cond(),
	m_is_started(false) {
	start();
}

thread_pool::~thread_pool() {
	if (m_is_started) {
		clear_task();
		stop();
	}
}

void thread_pool::stop() {
	{
		std::unique_lock<std::mutex> lock(m_mutex);
		m_is_started = false;
		m_cond.notify_all();
	}
	for (threads_t::iterator it = m_threads.begin(); it != m_threads.end(); ++it) {
		(*it)->join();
		delete *it;
	}

}

void thread_pool::add_task(const task_t& task) {
	std::unique_lock<std::mutex> lock(m_mutex);

	m_tasks.push_back(task);
	m_cond.notify_one();
}

void thread_pool::clear_task() {
	std::unique_lock<std::mutex> lock(m_mutex);
	m_tasks.clear();
}

std::size_t thread_pool::unactive_task() {
	std::unique_lock<std::mutex> lock(m_mutex);
	return m_tasks.size();
}

void thread_pool::joinAll()
{
	m_cond.notify_all();
	for(threads_t::iterator it = m_threads.begin(); it != m_threads.end(); ++it)
	{
		if ((*it)->joinable())
		{
			(*it)->join();
			delete *it;
		}
		m_cond.notify_one();
	}
	m_threads.clear();
}

bool thread_pool::is_started() {
	return m_is_started;
}

void thread_pool::start() {
	assert(m_threads.empty());
	m_is_started = true;
	m_threads.reserve(m_init_threads_size);
	for (int i = 0; i < m_init_threads_size; ++i) {
		m_threads.push_back(new std::thread(std::bind(&thread_pool::thread_loop, this)));
	}
}

void thread_pool::thread_loop() {
	while (m_is_started) {
		task_t task = take();
		if (task.lpThreadProc) {
//			task();

			task.lpThreadProc(&task);
		}
	}
}

task_t thread_pool::take() {
	std::unique_lock<std::mutex> lock(m_mutex);
	while (m_tasks.empty() && m_is_started) {
		m_cond.wait(lock);
	}
	task_t task;
	tasks_t::size_type size = m_tasks.size();
	if (!m_tasks.empty() && m_is_started) {
		task = m_tasks.front();
		m_tasks.pop_front();
		assert(size - 1 == m_tasks.size());
	}
	return task;
}